package org.hsqldb.index;

import org.hsqldb.HsqlNameManager.HsqlName;
import org.hsqldb.Row;
import org.hsqldb.RowBTree;
import org.hsqldb.RowBTreeDisk;
import org.hsqldb.Session;
import org.hsqldb.TableBase;
import org.hsqldb.persist.PersistentStore;
import org.hsqldb.persist.RowStoreBTreeDisk;
import org.hsqldb.types.Type;

public class IndexBTreeDisk extends IndexBTreeMemory{

	public IndexBTreeDisk(HsqlName name, long id, TableBase table,
			int[] columns, boolean[] descending, boolean[] nullsLast,
			Type[] colTypes, boolean pk, boolean unique, boolean constraint,
			boolean forward) {
		super(name, id, table, columns, descending, nullsLast, colTypes, pk, unique,
				constraint, forward);
	}
	
	public void insert(Session session, PersistentStore store, Row row){
		NodeBTreeBranchDisk n;
        int     compare      = -1;
        boolean compareRowId = !isUnique || hasNulls(session, row.getData());
        
        writeLock.lock();
        store.writeLock();
        
        try {
			// get keys
			Object[] keys = row.getData();
			Object[] newKeys = new Object[colTypes.length];

			for (int k = 0; k < colTypes.length; k++){
				newKeys[k] = keys[colIndex[k]];
			}
			// get root
			n = (NodeBTreeBranchDisk)getAccessor(store);
			
			// no root
			if (n == null){
				n = new NodeBTreeBranchDisk(table,position,colIndex);
				store.add(n);
				n.insertDataAtLeaf(store,newKeys, ((RowBTree)row).getNode(position), 0);
				store.setAccessor(this, n);
				
				return;
			}
			
			// search in tree
			while (true){
				int len = n.getChildNumber();
				int pos = 0;
				for (pos = 0; pos < len; pos++){
			        if (compareRowId) {
			            compare=row.getPos()-n.getChildPos(pos);
			        }
			        else{
			        	compare = compareKeys(session, row, n.getKeys(pos));
			        }
					if (compare <= 0){
						break;
					}
				}
				if (n.isLeaf){
					// already exists
					if (compare == 0){
						n.insertDulplicateData(store,((RowBTree)row).getNode(position), pos);
					} else{
						n.insertDataAtLeaf(store,newKeys, ((RowBTree)row).getNode(position), pos);
					}
					
					break;
				}
				n = (NodeBTreeBranchDisk)n.getChild(store, pos);
			}
			
			// check child number
			while (n.isOver()){
				NodeBTreeBranchDisk parentBranch = (NodeBTreeBranchDisk)n.getParent(store);
				
				NodeBTreeBranchDisk anotherBranch;
				if(n.isLeaf())
					anotherBranch = (NodeBTreeBranchDisk) n.split(store,n.getChildNumber() / 2);
				else {
					anotherBranch = (NodeBTreeBranchDisk) n.split(store,n.getChildNumber() / 2 + 1);
				}
				// is root
				if (parentBranch == null){
					NodeBTreeBranchDisk rootBranch = new NodeBTreeBranchDisk(table,position,colIndex);
					store.add(rootBranch);
					rootBranch.setIsLeaf(false);
					rootBranch.setChild(store,n, 0);
					n.setParent(store,rootBranch,0);
					rootBranch.setChild(store,anotherBranch, 1);
					anotherBranch.setParent(store,rootBranch,1);
					if(n.isLeaf()){
						n.setNextLeaf(store,anotherBranch);
						anotherBranch.setPreLeaf(store,n);
					}
					rootBranch.keyNumber++;
					if(n.isLeaf()){
						rootBranch.setKeys(anotherBranch.getKeys(0), 0);
					}
					else {
						rootBranch.setKeys(n.getKeys(n.getChildNumber() - 1), 0);
						n.setChild(store,null, n.keyNumber);
						n.setKeys(null, n.keyNumber - 1);
						n.keyNumber--;
					}
					rootBranch.setIsLeaf(false);
					store.setAccessor(this, rootBranch);
					break;
				}
				// normal node
				if (n.isLeaf()){
					anotherBranch.setNextLeaf(store,n.getNextLeaf(store));
					anotherBranch.setPreLeaf(store,n);
					if(n.getNextLeaf(store) != null){
						((NodeBTreeBranchDisk)(n.getNextLeaf(store))).setPreLeaf(store,anotherBranch);
					}
					n.setNextLeaf(store,anotherBranch);
					parentBranch.insertIntoNonLeaf(store,anotherBranch.getKeys(0), 
							anotherBranch, n.getPositionInParent());
					parentBranch.setChild(store,n, n.getPositionInParent());
				}
				else {
					parentBranch.insertIntoNonLeaf(store,n.getKeys(n.getChildNumber() - 1), anotherBranch, n.getPositionInParent());
					parentBranch.setChild(store,n, n.getPositionInParent());
					//FIXME:This should be completed by a method
					n.setChild(store,null, n.getChildNumber());
					n.setKeys(null, n.getChildNumber() - 1);
					n.keyNumber--;
				}
				
				anotherBranch.setParent(store,parentBranch,n.getPositionInParent() + 1);
				
				n = parentBranch;
				if(n == null)
					break;
			}
		} finally {
            store.writeUnlock();
            writeLock.unlock();
        }
	}
}
